package com.fable.common.socket;

import com.alibaba.fastjson2.JSONObject;
import com.fable.common.config.RabbitConfig;
import com.fable.common.constants.CommonConstant;
import com.fable.ntzw_ga.entity.ChatMsg;
import com.fable.ntzw_ga.service.IChatMsgService;
import com.rabbitmq.client.Channel;
import io.jsonwebtoken.Jwts;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Created by gzh on 2023/3/20 18:52
 */
@ServerEndpoint(value ="/websocket/{token}")
@Component
@Slf4j
public class WebSocketServer {

    @Autowired
    IChatMsgService chatMsgService;

    /**
     *静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
     */
    private static int onlineCount = 0;
    /**
     *concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
     */
    private static final ConcurrentHashMap<String, WebSocketServer> WEB_SOCKET_MAP = new ConcurrentHashMap<>();
    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    private Session session;

    /**
     * 密钥签名
     */
    private static String secret;

    @Value("${jwt.secret}")
    public void setSecret(String secret) {
        WebSocketServer.secret = secret;
    }

    /**
     * 消息发送者
     */
    private static String from = "";

    /**
     * 连接建立成功调用的方法*/
    @OnOpen
    public void onOpen(Session session,@PathParam("token") String token) {

        if(token!=null){
            String username = getUsername(token);
            //获取到session和username
            this.session = session;
            from = username;
            //username，则移除，再添加一个相同的username和新的Session（防止session过期）
            if(WEB_SOCKET_MAP.containsKey(username)){
                WEB_SOCKET_MAP.remove(username);
                WEB_SOCKET_MAP.put(username,this);
            }else{
                WEB_SOCKET_MAP.put(username,this);
                addOnlineCount();

            }
            log.info("用户连接:"+username+",当前在线人数为:" + getOnlineCount());
        }else{
            log.error("用户:"+from+",没有权限!!!!!!");
        }
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        //从set中删除
        //如果存在userId则移除，然后人数-1
        if(WEB_SOCKET_MAP.containsKey(from)){
            WEB_SOCKET_MAP.remove(from);
            //从set中删除
            subOnlineCount();
        }
        log.info("用户退出:"+from+",当前在线人数为:" + getOnlineCount());
    }

    /**
     * 收到客户端消息后调用的方法
     * @param message 客户端发送过来的消息
     *
     * */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("收到来自窗口"+from+"的信息:"+message);
        session.getAsyncRemote().sendText(JSONObject.toJSONString(new ChatMsg()));
    }

    /**
     *
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("发生错误");
        error.printStackTrace();
    }
    /**
     * 实现服务器主动推送
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    /**
     * 监听者,自定义推送消息
     * */
    public static void sendMqMessage(String message) throws IOException {
        ChatMsg chatMsg = JSONObject.parseObject(message, ChatMsg.class);
        if(chatMsg != null && !chatMsg.getFromUsername().equals(from)){
            if(chatMsg.getToUsername() != null && WEB_SOCKET_MAP.containsKey(chatMsg.getToUsername())){
                log.info("监听者收到"+chatMsg.getFromUsername()+"发来的消息，推送内容:"+chatMsg.getContent());
                WEB_SOCKET_MAP.get(chatMsg.getToUsername()).sendMessage(message);
            }
        }
    }

    /**
     * 从rabbitMq监听到消息，发送给当前登录用户
     * */
    @RabbitListener(queues = {RabbitConfig.DEFAULT_BOOK_QUEUE})
    public void listenerAutoAck(String text, Message message, Channel channel) {

        //  如果手动ACK,消息会被监听消费,但是消息在队列中依旧存在,如果 未配置 acknowledge-mode 默认是会在消费完毕后自动ACK掉
        final long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
//            log.info("text:{}",text);
            ChatMsg chatMsg = JSONObject.parseObject(text, ChatMsg.class);
            if (chatMsg!=null){
                if ( StringUtils.equals(chatMsg.getFromUsername(),from)){
                    log.info("过滤本地监听，重新压入MQ");
                    channel.basicRecover();
                } else {
                    chatMsg.setDir(CommonConstant.LEFT);
                    chatMsg.setFromNickname(chatMsg.getFromUsername());
                    chatMsgService.save(chatMsg);
                    sendMqMessage(text);
                    //  通知 MQ 消息已被成功消费,可以ACK了
                    channel.basicAck(deliveryTag, false);
                }
            }

        } catch (IOException e) {
            try {
                //  处理失败,重新压入MQ
                channel.basicRecover();
            } catch (IOException e1) {
                e1.printStackTrace();
            }
        }
    }
    /**
     * 群发自定义消息
     * */
    public static void sendInfo(String message,@PathParam("from") String from) throws IOException {
        JSONObject jsonObject = new JSONObject();
        log.info("推送消息到窗口"+from+"，推送内容:"+message);
    }

    /**
     * 解析token,从token中获取用户名
     */
    public String getUsername(String token) {
        return Jwts.parser().setSigningKey(secret).parseClaimsJws(token).getBody().getSubject();
    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    }
}
